package com.atguigu.wc

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._

/**
 * @description: flink流处理案例,flink每一个job或步骤都可以设置并行度,并行度不同的时候会根据re-balance来顺序分发
 * @time: 2020/6/17 23:27
 * @author: baojinlong
 **/
object StreamWordCount {
  def main(args: Array[String]): Unit = {
    // 创建流处理执行环境,一个任务就需要一个slot.如果并行的时候就需要两个
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 默认并行度是电脑核数,设置并行度,单个算子优先级>全局>任务提交配置的并行度
    // environment.setParallelism(8)
    // 来指示该operator不参与chaining（不会与前后的operator chain一起）
    // environment.disableOperatorChaining()
    // 从程序运行参数中读取hostname and port
    val params: ParameterTool = ParameterTool.fromArgs(args)
    println(params)
    val hostName: String = params.get("host")
    val port: Int = params.getInt("port")

    /**
     * 注意slotSharingGroup,disableChaining,startNewChain都是优化手段
     */
    // 接收socket文件流
    val inputDataStream: DataStream[String] = environment.socketTextStream(hostName, port)
    // 定义转换操作,设置slot共享组后不同编号的需要在不同的slot里面,并行的走另外一个slot,如flatMap使用2slot,并行度2
    val resultDataSet: DataStream[(String, Int)] = inputDataStream.flatMap(_.split(" ")).slotSharingGroup("1")
      .filter(_.nonEmpty).slotSharingGroup("2")
      // 设置disableChaining后任务会断开打散,也可以全局设置
      .map((_, 1)).disableChaining() // 转换成(word,count)元组
      .keyBy(0) // 根据第一个元素分组
      // startNewChain重新开启调用链
      .sum(1).startNewChain() // 按照第二个元素求和
    resultDataSet.print.setParallelism(1)
    // 正在等待事件来触发驱动
    environment.execute("stream word count")
  }
}
